package com.atguigu.dga.dolphinscheduler.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.dga.dolphinscheduler.bean.TDsTaskDefinition;
import com.atguigu.dga.dolphinscheduler.mapper.TDsTaskDefinitionMapper;
import com.atguigu.dga.dolphinscheduler.service.TDsTaskDefinitionService;
import com.atguigu.dga.meta.util.SqlUtil;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * <p>
 *  服务实现类
 * </p>
 *
 * @author zhangjunyi
 * @since 2023-06-09
 */
@Service
@DS("ds")
public class TDsTaskDefinitionServiceImpl extends ServiceImpl<TDsTaskDefinitionMapper, TDsTaskDefinition> implements TDsTaskDefinitionService {

    /**
     *
     * @param assessFullTableNameList  要考评表的所有表的全表名 库名.表名
     * @param assessDate 考评日期
     * @return
     */
    @Override
    public Map<String, TDsTaskDefinition> getTableDefinitionMapByDt(List<String> assessFullTableNameList, String assessDate) {
        //1.获取要考评表的所有定义信息  其次获取在考评日运行过的任务定义信息  并注意sql注入的问题 主要出现在拼接sql语句的时候
        List<TDsTaskDefinition> dsTaskDefinitionList = list(new QueryWrapper<TDsTaskDefinition>().in("name", assessFullTableNameList).exists("select 1 from t_ds_task_instance ti where ti.task_code=code and DATE_FORMAT(start_time,'%Y-%m-%d')='" + SqlUtil.filterUnsafeSql(assessDate)+"'"));

        HashMap<String, TDsTaskDefinition> taskDefinitionHashMap = new HashMap<>();

        for (TDsTaskDefinition tDsTaskDefinition : dsTaskDefinitionList) {
            //2.给sql属性赋值(提取任务参数中的sql，然后赋值给TDsTaskDefinition bean 中sql这个属性)
            String taskParams = tDsTaskDefinition.getTaskParams();
            String sql = getParamSql(taskParams);
            tDsTaskDefinition.setSql(sql);
            taskDefinitionHashMap.put(tDsTaskDefinition.getName(), tDsTaskDefinition);
        }
        return taskDefinitionHashMap;
    }

    private static String getParamSql(String taskParams){
        //1.先将json字符串转为JSONObject
        JSONObject jsonObject = JSON.parseObject(taskParams);
        String rawScript = jsonObject.getString("rawScript");

        //rawScript  #!/bin/bash set .....  with t1  insert .... from .... select ...... ;" hive -e.....

        Integer beginIdx = -1;
        Integer endIdx = -1;

        beginIdx = rawScript.indexOf("with");

        if (beginIdx==-1){
            beginIdx = rawScript.indexOf("insert");
        }

        int quoteIdx = rawScript.indexOf("\"", beginIdx);
        int semicolonsIdx = rawScript.indexOf(";", beginIdx);

        //首先判断sql结尾是否都有这两个符号
        if (quoteIdx>0&&semicolonsIdx>0){
            endIdx = Math.min(quoteIdx, semicolonsIdx);  //取下标最小的
        }else {
            endIdx = Math.max(quoteIdx, semicolonsIdx);//有一个符号没有的话 那么它所对应的下标是-1 因此直接取这两个符号下标最大的那一个
        }

        String sql = rawScript.substring(beginIdx, endIdx);

        return sql;
    }

    //测试
    public static void main(String[] args) {
        String taskParam = "{\"resourceList\":[],\"localParams\":[],\"rawScript\":\"#!/bin/bash\\n\\nsql=\\\"\\ninsert overwrite table gmall.dim_coupon_full partition(dt='${do_date}')\\nselect\\n    id,\\n    coupon_name,\\n    coupon_type,\\n    coupon_dic.dic_name,\\n    condition_amount,\\n    condition_num,\\n    activity_id,\\n    benefit_amount,\\n    benefit_discount,\\n    case coupon_type\\n        when '3201' then concat('满',condition_amount,'元减',benefit_amount,'元')\\n        when '3202' then concat('满',condition_num,'件打',10*(1-benefit_discount),'折')\\n        when '3203' then concat('减',benefit_amount,'元')\\n    end benefit_rule,\\n    create_time,\\n    range_type,\\n    range_dic.dic_name,\\n    limit_num,\\n    taken_count,\\n    start_time,\\n    end_time,\\n    operate_time,\\n    expire_time\\nfrom\\n(\\n    select\\n        id,\\n        coupon_name,\\n        coupon_type,\\n        condition_amount,\\n        condition_num,\\n        activity_id,\\n        benefit_amount,\\n        benefit_discount,\\n        create_time,\\n        range_type,\\n        limit_num,\\n        taken_count,\\n        start_time,\\n        end_time,\\n        operate_time,\\n        expire_time\\n    from gmall.ods_coupon_info_full\\n    where dt='${do_date}'\\n)ci\\nleft join\\n(\\n    select\\n        dic_code,\\n        dic_name\\n    from gmall.ods_base_dic_full\\n    where dt='${do_date}'\\n    and parent_code='32'\\n)coupon_dic\\non ci.coupon_type=coupon_dic.dic_code\\nleft join\\n(\\n    select\\n        dic_code,\\n        dic_name\\n    from gmall.ods_base_dic_full\\n    where dt='${do_date}'\\n    and parent_code='33'\\n)range_dic\\non ci.range_type=range_dic.dic_code;\\n\\\"\\n\\nhive -e \\\"$sql\\\"\",\"dependence\":{},\"conditionResult\":{\"successNode\":[],\"failedNode\":[]},\"waitStartTimeout\":{},\"switchResult\":{}}";
        String sql = getParamSql(taskParam);
        System.out.println(sql);

    }
}
